package lei.m_线程池_事物;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 优化版：解决总余额不一致问题的多线程转账示例
 */
public class OptimizedThreadPoolBalanceConsistencyExampleV1 {

    // 优化的线程池配置
    private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors();
    private static final int MAX_POOL_SIZE = CORE_POOL_SIZE * 2;
    private static final int QUEUE_CAPACITY = 50;
    private static final long KEEP_ALIVE_TIME = 30L;

    // 优化的线程池
    private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(
            CORE_POOL_SIZE,
            MAX_POOL_SIZE,
            KEEP_ALIVE_TIME,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(QUEUE_CAPACITY),
            new ThreadFactory() {
                private final AtomicInteger threadNumber = new AtomicInteger(1);

                @Override
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r, "TransferThread-" + threadNumber.getAndIncrement());
                    t.setDaemon(false);
                    return t;
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    // 预热线程池
    static {
        executor.prestartAllCoreThreads();
    }

    // 事务上下文传递
    private static final ThreadLocal<TransactionContext> transactionContext =
            new ThreadLocal<>();

    // 原子操作的银行账户 - 使用CAS操作保证原子性
    static class AtomicBankAccount {
        private final String accountId;
        private final AtomicLong balance;

        public AtomicBankAccount(String accountId, long initialBalance) {
            this.accountId = accountId;
            this.balance = new AtomicLong(initialBalance);
        }

        /**
         * 原子转账操作 - 保证总余额不变
         * 使用单一CAS操作完成转账，避免中间状态
         */
        public boolean transferTo(AtomicBankAccount toAccount, long amount) {
            // 检查余额（快速失败）
            if (this.balance.get() < amount) {
                return false;
            }

            // 使用原子操作进行转账
            long fromBalance = this.balance.get();
            long toBalance = toAccount.balance.get();

            // 确保余额足够
            if (fromBalance < amount) {
                return false;
            }

            // 原子性地更新两个账户的余额
            if (this.balance.compareAndSet(fromBalance, fromBalance - amount)) {
                // 转出成功，更新转入账户
                toAccount.balance.addAndGet(amount);
                return true;
            }

            return false; // CAS失败，重试或失败
        }

        // 带重试机制的转账
        public boolean transferToWithRetry(AtomicBankAccount toAccount, long amount, int maxRetries) {
            for (int i = 0; i < maxRetries; i++) {
                if (transferTo(toAccount, amount)) {
                    return true;
                }
                // 短暂等待后重试
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return false;
                }
            }
            return false;
        }

        public long getBalance() {
            return balance.get();
        }

        public String getAccountId() {
            return accountId;
        }
    }

    static class TransactionContext {
        private boolean inTransaction = false;
        private String transactionId;
        private long startTime;

        public TransactionContext(String transactionId) {
            this.transactionId = transactionId;
            this.startTime = System.currentTimeMillis();
        }

        // getters and setters
        public boolean isInTransaction() { return inTransaction; }
        public void setInTransaction(boolean inTransaction) { this.inTransaction = inTransaction; }
        public String getTransactionId() { return transactionId; }
        public void setTransactionId(String transactionId) { this.transactionId = transactionId; }
        public long getStartTime() { return startTime; }
    }

    // 共享的银行账户 - 使用原子操作保证一致性
    private static final AtomicBankAccount accountA = new AtomicBankAccount("ACCOUNT_A", 10000L);
    private static final AtomicBankAccount accountB = new AtomicBankAccount("ACCOUNT_B", 10000L);
    private static final AtomicBankAccount accountC = new AtomicBankAccount("ACCOUNT_C", 10000L);

    // 初始总余额
    private static final long INITIAL_TOTAL_BALANCE =
            accountA.getBalance() + accountB.getBalance() + accountC.getBalance();

    public static void main(String[] args) {
        System.out.println("=== 优化版多线程转账示例（保证总余额一致性） ===");
        System.out.println("初始余额: A=" + accountA.getBalance() +
                ", B=" + accountB.getBalance() +
                ", C=" + accountC.getBalance());
        System.out.println("初始总余额: " + INITIAL_TOTAL_BALANCE);

        long startTime = System.currentTimeMillis();

        // 使用CountDownLatch确保所有任务完成
        CountDownLatch latch = new CountDownLatch(12);

        // 提交任务
        for (int i = 0; i < 12; i++) {
            final int taskId = i;
            executor.submit(() -> {
                try {
                    // 设置事务上下文
                    TransactionContext context = new TransactionContext("TX-" + taskId);
                    context.setInTransaction(true);
                    transactionContext.set(context);

                    // 执行转账
                    performAtomicTransfer(taskId);

                } catch (Exception e) {
                    System.err.println("任务执行失败: " + e.getMessage());
                } finally {
                    // 清理事务上下文
                    transactionContext.remove();
                    // 计数器减一
                    latch.countDown();
                }
            });
        }

        // 等待所有任务完成
        try {
            if (latch.await(10, TimeUnit.SECONDS)) {
                System.out.println("所有任务执行完成");
            } else {
                System.out.println("任务执行超时");
            }
        } catch (InterruptedException e) {
            System.err.println("等待任务完成时被中断");
            Thread.currentThread().interrupt();
        }

        // 优雅关闭线程池
        shutdownExecutor();

        long endTime = System.currentTimeMillis();

        // 最终结果
        long finalBalanceA = accountA.getBalance();
        long finalBalanceB = accountB.getBalance();
        long finalBalanceC = accountC.getBalance();
        long finalTotalBalance = finalBalanceA + finalBalanceB + finalBalanceC;

        System.out.println("=== 最终余额 ===");
        System.out.println("账户A余额: " + finalBalanceA);
        System.out.println("账户B余额: " + finalBalanceB);
        System.out.println("账户C余额: " + finalBalanceC);
        System.out.println("最终总余额: " + finalTotalBalance);
        System.out.println("余额一致性检查: " +
                (finalTotalBalance == INITIAL_TOTAL_BALANCE ? "通过" : "失败"));
        System.out.println("总执行时间: " + (endTime - startTime) + " ms");
    }

    // 原子转账操作
    private static void performAtomicTransfer(int taskId) {
        boolean success = false;
        long amount = 50L + (taskId % 10) * 10L; // 不同金额

        switch (taskId % 3) {
            case 0:
                success = accountA.transferToWithRetry(accountB, amount, 3);
                break;
            case 1:
                success = accountB.transferToWithRetry(accountC, amount, 3);
                break;
            case 2:
                success = accountC.transferToWithRetry(accountA, amount, 3);
                break;
        }

        if (success) {
            System.out.println("任务 " + taskId + " 转账成功: " + amount +
                    ", 线程: " + Thread.currentThread().getName());
        } else {
            System.out.println("任务 " + taskId + " 转账失败: " + amount +
                    ", 线程: " + Thread.currentThread().getName());
        }
    }

    private static void shutdownExecutor() {
        executor.shutdown();
        try {
            if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                System.out.println("线程池关闭超时，强制关闭");
                executor.shutdownNow();
                if (!executor.awaitTermination(2, TimeUnit.SECONDS)) {
                    System.err.println("线程池无法正常关闭");
                }
            }
        } catch (InterruptedException e) {
            System.err.println("线程池关闭被中断");
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
        System.out.println("线程池已关闭");
    }
}
